package com.competition.web.admin.queue;

import java.util.HashMap;
import java.util.Map;

import javax.annotation.PreDestroy;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import com.competition.web.admin.queue.dispatch.IBlockchainOperation;

/**
 * 队列消费者
 * 
 * @author 111
 */
@Component
public class SysMsgQueueCosumer<T> implements Runnable {

	private static Logger logger = LoggerFactory.getLogger(SysMsgQueueCosumer.class);

	@Autowired
	protected StringRedisTemplate stringRedisTemplate;

	@Autowired
	private SysMsgQueue<T> sysMsgQueue;

	private static Map<Integer, IBlockchainOperation> map = new HashMap<>();

	static {
//		try {
//			map.put(0, (IBlockchainOperation) Class.forName("com.fingerstar.manager.queue.dispatch.impl.Purchase")
//					.newInstance());
//			map.put(1, (IBlockchainOperation) Class.forName("com.fingerstar.manager.queue.dispatch.impl.Casting")
//					.newInstance());
//		} catch (InstantiationException e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		} catch (IllegalAccessException e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		} catch (ClassNotFoundException e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		}

	}

//	@Autowired
//	private IProductdetailsService productdetailsService;
//	
//	@Autowired
//	private ISystemConfig systemConfig;
//	
//	@Autowired
//	private IOrderGoodsService orderGoodsService;

	private ThreadPoolTaskExecutor threadPoolTaskExecutor;

	@Autowired
	public SysMsgQueueCosumer(ThreadPoolTaskExecutor threadPoolTaskExecutor) {

		this.threadPoolTaskExecutor = threadPoolTaskExecutor;

		this.init();
	}

//	@PostConstruct
	public void init() {
		
//		new Thread(this, "MsgCosumer").start();
	}

	@Override
	public void run() {
		logger.info("SysMsgQueueCosumer:");
		// 阻塞，等待生产者生产数据，继续执行
		try {
			MsgQueueConstant.cosumerSemaphore.acquire();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		while (true) {
			try {
				execute();
			} catch (Exception e) {
				e.printStackTrace();
				logger.info("消费者发生异常:{}", e.getMessage());
			}
		}
	}

	@PreDestroy
	public void destory() {
		this.threadPoolTaskExecutor.shutdown();
	}

	/**
	 * 消费消息
	 */
	private void execute() {
		T msg = sysMsgQueue.poll();

		if (null == msg) {
			// 获取队列消息，如消息为null，则释放lock锁，继续队列添加消息
			// 队列中执行完毕，通知生产者生产数据
			logger.info("消费者：队列中执行完毕，通知生产者生产数据");
			MsgQueueConstant.producerSemaphore.release();
			try {
				logger.info("消费者：队列中执行完毕，等待生产者生产数据");
				// 阻塞，等待生产者生产数据，继续执行
				MsgQueueConstant.cosumerSemaphore.acquire();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			return;
		}

		logger.info("--------开始消费消息{}", msg);

		this.threadPoolTaskExecutor.execute(() -> {
			try {
				int function = (int) msg.getClass().getMethod("getPayFunction").invoke(msg);

				map.get(function).operation(msg);

			} catch (Exception e) {
				logger.debug("错误消息：{}", e);
			}
		});
	}

}
